package consumer_manager

import (
	"context"
	"fmt"
	"net/http"
	"sync"
	"time"

	"gitee.com/Sxiaobai/gs/gsdb"
	"gitee.com/Sxiaobai/gs/gstool"
	"github.com/gin-gonic/gin"
)

// Manager 消费者管理器
type Manager struct {
	config     *Config
	nsqAdapter NSQConsumerInterface
	dbManager  *DatabaseManager
	logger     *gstool.GsSlog
	callFunc   func(string, uint16) bool

	// API服务器管理
	apiServer  *APIServer
	httpServer *http.Server

	// 内部状态
	mu           sync.RWMutex
	ctx          context.Context
	cancel       context.CancelFunc
	reportTicker *time.Ticker
}

// NewManager 创建管理器
func NewManager(config *Config, nsqAdapter NSQConsumerInterface, db *gsdb.GsMysql, logger *gstool.GsSlog) (*Manager, error) {
	if err := config.Validate(); err != nil {
		return nil, fmt.Errorf("配置验证失败: %v", err)
	}

	ctx, cancel := context.WithCancel(context.Background())

	dbManager := NewDatabaseManager(db, config.TableName, logger)

	return &Manager{
		config:     config,
		nsqAdapter: nsqAdapter,
		dbManager:  dbManager,
		logger:     logger,
		ctx:        ctx,
		cancel:     cancel,
	}, nil
}

// SetCallFunc 设置消费回调函数
func (m *Manager) SetCallFunc(callFunc func(string, uint16) bool) {
	m.callFunc = callFunc
}

// Start 启动管理器
func (m *Manager) Start() error {
	if m.callFunc == nil {
		return fmt.Errorf("消费回调函数未设置")
	}

	// // 创建数据库表
	// if err := m.dbManager.CreateTable(); err != nil {
	// 	return fmt.Errorf("创建数据库表失败: %v", err)
	// }

	// 启动定时上报
	m.startReporting()

	m.logger.Infof("消费者管理器启动成功: %s@%s", m.config.ConsumerName, m.config.HostIP)
	return nil
}

// StartAPIServer 启动API服务器
func (m *Manager) StartAPIServer() error {
	if !m.config.HTTPEnabled {
		m.logger.Infof("HTTP API服务器已禁用")
		return nil
	}

	// 创建API服务器
	m.apiServer = NewAPIServer(m, m.config, m.logger)
	m.apiServer.engine = gin.Default()
	m.apiServer.SetupRoutes(m.apiServer.engine)

	// 创建HTTP服务器
	m.httpServer = &http.Server{
		Addr:    ":" + m.config.HTTPPort,
		Handler: m.apiServer.engine,
	}

	// 在goroutine中启动服务器
	go func() {
		m.logger.Infof("HTTP API服务器启动在端口: %s", m.config.HTTPPort)
		if err := m.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			m.logger.Errof("HTTP API服务器启动失败: %v", err)
		}
	}()

	return nil
}

// Stop 停止管理器 - 实现完整的优雅停止逻辑
func (m *Manager) Stop() error {
	m.logger.Infof("开始停止消费者管理器...")

	// 1. 取消上下文，停止所有goroutine
	m.cancel()

	// 2. 停止定时上报
	if m.reportTicker != nil {
		m.reportTicker.Stop()
		m.logger.Infof("定时上报已停止")
	}

	// 3. 优雅停止HTTP API服务器
	if m.httpServer != nil {
		m.logger.Infof("正在停止HTTP API服务器...")
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()

		if err := m.httpServer.Shutdown(ctx); err != nil {
			m.logger.Errof("HTTP API服务器停止失败: %v", err)
		} else {
			m.logger.Infof("HTTP API服务器已优雅停止")
		}
	}

	// 4. 停止NSQ消费者和生产者
	if m.nsqAdapter != nil {
		m.logger.Infof("正在停止NSQ消费者...")
		currentCount := m.nsqAdapter.GetCurrentConsumerCount()
		if currentCount > 0 {
			if err := m.nsqAdapter.ForceStopAllConsumers(); err != nil {
				m.logger.Errof("停止NSQ消费者失败: %v", err)
			} else {
				m.logger.Infof("NSQ消费者已停止，停止数量: %d", currentCount)
			}
		}
		// 停止生产者
		m.nsqAdapter.StopProducer()
	}

	// 5. 最后上报一次状态
	m.logger.Infof("上报最终状态...")
	m.reportStatus()

	m.logger.Infof("消费者管理器已完全停止")
	return nil
}

// GetCurrentCount 获取当前消费者数量
func (m *Manager) GetCurrentCount() int {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.nsqAdapter.GetCurrentConsumerCount()
}

// AdjustCount 调整消费者数量
func (m *Manager) AdjustCount(targetCount int) error {
	if targetCount < m.config.MinConsumers {
		return fmt.Errorf("消费者数量不能少于%d个", m.config.MinConsumers)
	}
	if targetCount > m.config.MaxConsumers {
		return fmt.Errorf("消费者数量不能超过%d个", m.config.MaxConsumers)
	}

	m.mu.Lock()
	defer m.mu.Unlock()

	currentCount := m.nsqAdapter.GetCurrentConsumerCount()

	if targetCount != currentCount {
		// 调整消费者数量
		if err := m.nsqAdapter.AdjustConsumerCount(targetCount, m.callFunc); err != nil {
			return fmt.Errorf("调整消费者数量失败: %v", err)
		}
		m.logger.Infof("消费者数量已调整: %d -> %d", currentCount, targetCount)

		// 立即上报状态
		go m.reportStatus()
	}

	return nil
}

// PauseConsumers 暂停所有消费者
func (m *Manager) PauseConsumers() error {
	m.mu.Lock()
	defer m.mu.Unlock()

	currentCount := m.nsqAdapter.GetCurrentConsumerCount()
	if currentCount == 0 {
		m.logger.Infof("消费者已经处于暂停状态")
		return nil
	}

	// 使用ForceStopAllConsumers方法停止所有消费者
	if err := m.nsqAdapter.ForceStopAllConsumers(); err != nil {
		return fmt.Errorf("暂停消费者失败: %v", err)
	}

	m.logger.Infof("消费者已暂停，停止数量: %d", currentCount)

	// 立即上报状态
	go m.reportStatus()

	return nil
}

// startReporting 启动定时上报
func (m *Manager) startReporting() {
	m.reportTicker = time.NewTicker(m.config.ReportInterval)
	go func() {
		// 立即上报一次
		m.reportStatus()

		for {
			select {
			case <-m.ctx.Done():
				return
			case <-m.reportTicker.C:
				m.reportStatus()
			}
		}
	}()
}

// reportStatus 上报状态
func (m *Manager) reportStatus() {
	currentCount := m.GetCurrentCount()
	
	// 处理spawn_consumer_count大于max_consumer_count的情况
	spawnCount := m.config.SpawnConsumerCount
	if spawnCount > m.config.MaxConsumers {
		spawnCount = m.config.MaxConsumers
	}
	
	err := m.dbManager.UpsertStatus(m.config.ConsumerName, m.config.HostIP, m.config.HostPort, m.config.NSQLookupHost, m.config.NSQHTTPAddress, m.config.NSQAdminAddress, m.config.Topic, m.config.Channel, m.config.ConsumerRemark, m.config.MaxConsumers, currentCount, spawnCount, m.config.DepthLimit)
	if err != nil {
		m.logger.Errof("上报消费者状态失败: %v", err)
	}
}

// GetStatus 获取状态信息
func (m *Manager) GetStatus() *StatusData {
	// 处理spawn_consumer_count大于max_consumer_count的情况
	spawnCount := m.config.SpawnConsumerCount
	if spawnCount > m.config.MaxConsumers {
		spawnCount = m.config.MaxConsumers
	}
	
	// 获取当前的scaling_mode状态
	status, err := m.dbManager.GetStatus(m.config.ConsumerName, m.config.HostIP, m.config.HostPort, m.config.Topic, m.config.Channel)
	scalingMode := "manual" // 默认值
	if err == nil {
		scalingMode = status.ScalingMode
	}
	
	return &StatusData{
		CurrentCount:       m.GetCurrentCount(),
		MaxConsumerCount:   m.config.MaxConsumers,
		SpawnConsumerCount: spawnCount,
		DepthLimit:         m.config.DepthLimit,
		ScalingMode:        scalingMode,
		ConsumerName:       m.config.ConsumerName,
		ServerIP:           m.config.HostIP,
		Port:               m.config.HostPort,
		NSQLookupHost:      m.config.NSQLookupHost,
		NSQHTTPAddress:     m.config.NSQHTTPAddress,
		NSQAdminAddress:    m.config.NSQAdminAddress,
		Topic:              m.config.Topic,
		Channel:            m.config.Channel,
		LastUpdate:         time.Now().Format("2006-01-02 15:04:05"),
		ConsumerRemark:     m.config.ConsumerRemark,
	}
}

// ToggleScalingMode 切换scaling_mode状态
func (m *Manager) ToggleScalingMode() error {
	return m.dbManager.ToggleScalingMode(m.config.ConsumerName, m.config.HostIP, m.config.HostPort, m.config.Topic, m.config.Channel)
}
